package com.atguigu.gmall.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/28 9:21
 */
public class FlinkSourceUtil {
    
    public static FlinkKafkaConsumer<String> getKafkaSource(String groupId, String topic) {
        Properties conf = new Properties();
        conf.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        conf.setProperty("group.id", groupId);
        // 如果启动的时候, 这个消费者对这个topic的消费没有上次的消费记录, 就从这个配置的位置开始消费
        // 如果有消费记录, 则从上次的位置开始消费
        conf.setProperty("auto.offset.reset", "latest");
        conf.setProperty("isolation.level", "read_committed");
        
        return new FlinkKafkaConsumer<>(
            topic,
            new SimpleStringSchema(),
            conf
        );
    }
}
